package com.tee.live.user.provider.config;

import com.alibaba.fastjson.JSON;
import com.tee.live.framework.redis.starter.key.UserProviderCacheKeyBuilder;
import com.tee.live.user.constans.CacheAsyncDeleteCode;
import com.tee.live.user.dto.UserCacheAsyncDeleteDTO;
import com.tee.live.user.dto.UserDTO;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.List;
import java.util.Map;

import static com.tee.live.user.constans.UserProviderTopicNames.CACHE_ASYNC_DELETE_TOPIC;

@Configuration
public class RocketMQConsumerConfig implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQConsumerConfig.class);
    @Resource
    private RocketMQConsumerProperties consumerProperties;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private UserProviderCacheKeyBuilder userProviderCacheKeyBuilder;



    @Override
    public void afterPropertiesSet() throws Exception {
        initConsumer();
    }

    private void initConsumer() {
        try {
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
            defaultMQPushConsumer.setNamesrvAddr(consumerProperties.getNameSrv());
            defaultMQPushConsumer.setConsumerGroup(consumerProperties.getGroupName());
            defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            defaultMQPushConsumer.subscribe(CACHE_ASYNC_DELETE_TOPIC, "");
            defaultMQPushConsumer.setMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> list,
                                                                final ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    String msgStr = new String(list.get(0).getBody());
                    UserCacheAsyncDeleteDTO userCacheAsyncDeleteDTO = JSON.parseObject(msgStr, UserCacheAsyncDeleteDTO.class);
                    if (userCacheAsyncDeleteDTO == null) {
                        logger.error("消息参数为null ,参数异常,内容:{}", msgStr);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    if (userCacheAsyncDeleteDTO.getCode() == CacheAsyncDeleteCode.USER_INFO_DELETE.getCode()) {
                        // 删除用户信息缓存
                        Map<String,Object> parse = JSON.parseObject(userCacheAsyncDeleteDTO.getJson());
                        Long userId = (Long) parse.get("userId");
                        redisTemplate.delete(userProviderCacheKeyBuilder.buildTagKey(userId));
                    } else if (userCacheAsyncDeleteDTO.getCode() == CacheAsyncDeleteCode.USER_TAG_DELETE.getCode()) {
                        // 删除用户标签缓存
                        Map<String,Object> parse = JSON.parseObject(userCacheAsyncDeleteDTO.getJson());
                        Long userId = (Long) parse.get("userId");
                        redisTemplate.delete(userProviderCacheKeyBuilder.buildTagKey(userId));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            defaultMQPushConsumer.start();
            logger.info("RocketMq消费者启动成功,nameSev:{}", consumerProperties.getNameSrv());
        } catch (MQClientException e) {
            throw new RuntimeException(e);
        }


    }

}
